package com.k8.common.event;

import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;


/**
 * 并发控制的效果是否好，取决于采取了sleep，但是最终执行完所有任务的时间开销与不采取sleep的开销大致相同
 * 此外，在短时间处理的任务和长时间处理的任务都要有较好的表现，因为任务执行时间长sleep的时间短则看不出来差距
 * 并发量高和低的效果都要好，并发量高的话横竖都有任务执行，sleep的时间只要不夸张都会有很好的效果，看不出来差距
 * <p>
 * 所以要在并发量低任务执行时间短的环境下测试
 */
@Slf4j
public class EventDeployer {
    //private Logger logger = LoggerFactory.getLogger(EventDeployer.class);
    private final Map<Class<? extends Listener>, List<Listener>> cacheListeners = new ConcurrentHashMap<>();
    private final Map<Class<? extends SynListener>, List<SynListener>> cacheSynListeners = new ConcurrentHashMap<>();

    private final ExecutorService executorService;
    //持有锁的时间
    private long baseSleepTime = 50;
    private long maxSleepTime = 1000;

    public EventDeployer(ExecutorService executorService) {
        this.executorService = executorService;
    }

    public EventDeployer(ExecutorService executorService, long baseSleepTime, long maxSleepTime) {
        this.executorService = executorService;
        this.baseSleepTime = baseSleepTime;
        this.maxSleepTime = maxSleepTime;
    }

    /**
     * Listener派生类型较少，可以使用一实现一锁的方式
     * 1.添加listener时该类型的事件发布会被阻塞
     */

    public void addListener(Listener listener) {
        if (listener == null) throw new IllegalArgumentException("Listener is null");
        if (listener instanceof SynListener synListener) {
            Class<? extends SynListener> clazzKey = synListener.getClass();
            List<SynListener> synListeners = cacheSynListeners.get(clazzKey);
            if (synListeners == null) {
                synListeners = cacheSynListeners.computeIfAbsent(clazzKey, key -> {
                    return Collections.synchronizedList(new LinkedList<>());
                });
            }
            if (synListener.getLock() == null) {
                throw new IllegalStateException("This is a synListener, but lock is null");
            }
            //线程安全list
            synListeners.add(synListener);
        } else {
            Class<? extends Listener> clazzKey = listener.getClass();
            List<Listener> listeners = cacheListeners.get(clazzKey);
            if (listeners == null) {
                listeners = cacheListeners.computeIfAbsent(clazzKey, key -> {
                    return Collections.synchronizedList(new LinkedList<>());
                });
            }
            listeners.add(listener);
        }
        log.info("Add a listener,: " + listener.getClass().getName());
    }

    public void publishEvent(Event event) {
        publishEvent(event, false);
    }

    public void publishEvent(Event event, boolean retry) {
        //对于没有线程同步需求的监听器，处理事件全部都可以使用异步去处理
        Set<Class<? extends Listener>> clazzKeys = cacheListeners.keySet();
        List<BiValue> canDisposalList = new LinkedList<>();
        CompletableFuture<Void> allOfFuture = null;
        for (Class<? extends Listener> clazzKey : clazzKeys) {
            List<Listener> listeners = cacheListeners.get(clazzKey);
            if (!listeners.isEmpty()) {
                Listener listener0 = listeners.get(0);
                if (listener0.judgeEventType(event)) {
                    for (Listener listener : listeners) {
                        if (listener.canDispose(event)) {
                            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                                listener.disposal(event);
                            }, executorService);
                            BiValue biValue = new BiValue(listener, future);
                            canDisposalList.add(biValue);
                            if (allOfFuture == null) {
                                allOfFuture = CompletableFuture.allOf(future);
                            } else {
                                allOfFuture = CompletableFuture.allOf(allOfFuture, future);
                            }
                        }
                    }
                }
            }
        }
        LinkedList<Listener> canRetryFailListeners = null;
        //发布事件给同步执行监听器
        LinkedList<SynListener> canRetryFailSynListeners = publishEventToSynListener(event);
        if (allOfFuture != null) {
            try {
                allOfFuture.get();
            } catch (InterruptedException | ExecutionException e) {
                canRetryFailListeners = canDisposalList.stream()
                        .filter(biValue -> {
                            try {
                                biValue.future.get();
                            } catch (InterruptedException | ExecutionException ex) {
                                log.warn("Listener disposal event failed , type of listener : " + biValue.listener.getClass().getName());
                                throw new RuntimeException(ex);
                            }
                            return false;
                        }).map(biValue -> {
                            return biValue.listener;
                        }).collect(Collectors.toCollection(LinkedList::new));
                //因为后续处理需要用LinkedList做队列存储结构，比ArrayList效率高
            }
        }
        //如果需要重试
        if (retry) {
            //流式操作因为可以是异步的，所以被判定非线程安全，会爆红
            LinkedList<Listener> finalCanRetryFailListeners = canRetryFailListeners;
            executorService.submit(() -> {
                retryPublishEvent(event, finalCanRetryFailListeners);
            });

            executorService.submit(() -> {
                retryPublishEventToSynListeners(event, canRetryFailSynListeners);
            });
        }
    }

    private void retryPublishEventToSynListeners(Event event, LinkedList<SynListener> canRetryFailSynListeners) {
        if (canRetryFailSynListeners.isEmpty()) return;
        while (!canRetryFailSynListeners.isEmpty()) {
            LinkedList<SynListener> newCanRetryFailSynListeners = new LinkedList<>();
            acquire(canRetryFailSynListeners, event, newCanRetryFailSynListeners);
            canRetryFailSynListeners = newCanRetryFailSynListeners;
        }
    }

    private void retryPublishEvent(Event event, LinkedList<Listener> canRetryFailListeners) {
        if (canRetryFailListeners == null || canRetryFailListeners.isEmpty()) return;
        while (!canRetryFailListeners.isEmpty()) {
            Listener listener = canRetryFailListeners.removeFirst();
            try {
                listener.disposal(event);
            } catch (Throwable throwable) {
                canRetryFailListeners.addLast(listener);
            }
        }
    }

    public Future<?> publishEventAsync(Event event) {
        return publishEventAsync(event, false);
    }

    public Future<?> publishEventAsync(Event event, boolean retry) {
        return executorService.submit(() -> {
            publishEvent(event, retry);
        });
    }

    /**
     * 不直接在事件监听器处实现同步是因为并没有给每一个监听器单独开一个线程去处理事件
     * 事件还是由发布事件的线程进行处理，如果某个线程拿到了一个监听器的锁，表示它在处理并阻塞其它线程
     * 那么同一时刻就只能有一个线程去处理一个事件，一个事件处理时间设为t，每个线程发布n个事件，同时l个监听器需要去处理
     * 时间开销为ntl
     * 当使用tryLock，没拿到锁则尝试发布给别的监听器，那么时间开销最优为为nt
     * 添加重试机制，由用户选择是否重试
     */
    private LinkedList<SynListener> publishEventToSynListener(Event event) {
        LinkedList<SynListener> canRetryFailSynListeners = new LinkedList<>();
        LinkedList<SynListener> synListeners = new LinkedList<>();
        Set<Class<? extends SynListener>> keySet = cacheSynListeners.keySet();
        for (Class<? extends SynListener> clazzKey : keySet) {
            List<SynListener> listeners = cacheSynListeners.get(clazzKey);
            if (!listeners.isEmpty()) {
                SynListener listener = listeners.get(0);
                if (listener.judgeEventType(event)) {
                    for (SynListener synListener : listeners) {
                        if (listener.canDispose(event)) {
                            Lock lock = synListener.getLock();
                            boolean lockSuccess = false;
                            //在遍历添加时先尝试获取一次
                            try {
                                lockSuccess = lock.tryLock();
                                if (lockSuccess) {
                                    synListener.disposal(event);
                                } else {
                                    synListeners.add(synListener);
                                }
                            } catch (Throwable throwable) {
                                canRetryFailSynListeners.add(synListener);
                            } finally {
                                if (lockSuccess) {
                                    lock.unlock();
                                }
                            }
                        }
                    }
                }
            }
        }
        acquire(synListeners, event, canRetryFailSynListeners);
        return canRetryFailSynListeners;
    }

    private void acquire(LinkedList<SynListener> synListeners, Event event, LinkedList<SynListener> canRetryFailSynListeners) {
        //执行轮次,到达n+i轮，且n+i轮还是没有抢到监听器锁，则睡眠，且使用指数避退，每睡眠一次n就除以2，直到n=16(maxBaseTimes)或synListeners.size()，
        byte postSpins = 0;
        byte spins = 0;
        long sleepTime = -1L;
        int sleepCount = 1;
        //开始争夺监听器执行
        while (!synListeners.isEmpty()) {
            //当执行n轮后，还能进入，说明并发量较高
            //如果这一轮拿到过锁，就继续执行
            //如果没有则睡眠，避免cpu资源浪费，当达到最大等待睡眠，则不再睡眠，避免线程饿死，让新来的线程睡眠
            boolean disposal = false;
            int size = synListeners.size();
            for (int i = 0; i < size; ++i) {
                SynListener synListener = synListeners.removeFirst();
                Lock lock = synListener.getLock();
                boolean lockSuccess = false;
                try {
                    //当spins小于0，已经等待很长时间了，说明并发度很高，直接拿锁，没拿到则会逐渐进入等待队列（在syn的acquire中还会有多次尝试）
                    if (spins < 0) {
                        lock.lock();
                        lockSuccess = true;
                    } else {
                        lockSuccess = lock.tryLock();
                    }
                    if (lockSuccess) {
                        disposal = true;
                        synListener.disposal(event);
                    } else {
                        synListeners.addLast(synListener);
                    }
                } catch (Throwable throwable) {
                    canRetryFailSynListeners.add(synListener);
                } finally {
                    if (lockSuccess) {
                        lock.unlock();
                    }
                }
            }
            //没有拿到过锁就应该睡眠，但是如果睡眠时间过长==maxSleepTime就不应该再睡眠
            if (!disposal) {
                if (spins > 0) {
                    spins--;
                    //在有并发的时候，Thread.onSpinWait()使线程空转反而使得总体开销比不加这步整体开销块，因为这里空转减少了抢锁的可能性
                    //使得由于并发没抢到锁而进入尝试睡眠的代码块的次数反而减少了，总体开销就降低了
                    //在低并发时提升明显，高并发且处理时间长时就没什么效果，因为大量的线程被阻塞会很快到达最大值127
                    Thread.onSpinWait();
                } else {
                    //指数避退增加睡眠时间
                    if (sleepTime < maxSleepTime) {
                        sleepTime = (int) Math.pow(2, sleepCount) * baseSleepTime - baseSleepTime;
                        sleepCount++;
                    }
                    if (sleepTime > maxSleepTime) {
                        sleepTime = maxSleepTime;
                    }
                    //到达127之后就会为-1
                    spins = postSpins = (byte) (postSpins << 1 | 1);
                    try {
                        Thread.sleep(sleepTime);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        }
    }

    private class BiValue {
        Listener listener;
        CompletableFuture<Void> future;

        public BiValue(Listener listener, CompletableFuture<Void> future) {
            this.listener = listener;
            this.future = future;
        }
    }
}
